from fastmcp import FastMCP  
import asyncio  
from fastmcp import Client  
import akshare as ak  
  
# 必须先初始化MCP实例  
mcp = FastMCP('news.mcp')  
  
@mcp.tool()  
def get_news() -> list:  
    """  
    获取静态财经新闻数据(fallback数据)  
      
    :result: 返回新闻列表,每个元素包含标题、摘要、来源和日期  
    """  
    return [  
        {  
            "title": "央行宣布降准0.5个百分点",  
            "summary": "人民银行决定下调金融机构存款准备金率0.5个百分点，释放长期资金约1万亿元。",  
            "source": "经济日报",  
            "date": "2023-11-20"  
        },  
        {  
            "title": "多部委联合发文支持民营经济发展",  
            "summary": "国家发改委等多部门联合发布支持民营经济发展的政策措施，涉及融资、税收等多方面。",  
            "source": "证券时报",  
            "date": "2023-11-18"  
        }  
    ]  
  
@mcp.tool()  
def get_news_from_akshare() -> list:  
    """  
    使用AKShare获取东方财富网财经新闻(官方API,最稳定)  
      
    :result: 返回新闻列表,每个元素包含标题、链接和发布时间  
    """  
    try:  
        # 使用AKShare的东方财富新闻接口  
        # symbol参数: "全部"表示获取所有新闻, 也可以指定具体股票代码  
        news_df = ak.stock_news_em(symbol="全部")  
          
        news_items = []  
        # 获取前10条新闻  
        for _, row in news_df.head(10).iterrows():  
            news_items.append({  
                "title": row.get('新闻标题', ''),  
                "url": row.get('新闻链接', ''),  
                "date": str(row.get('发布时间', '')),  
                "source": "东方财富网"  
            })  
          
        return news_items if news_items else [{"提示": "暂无新闻数据"}]  
    except Exception as e:  
        print(f"AKShare获取新闻失败: {e}")  
        # 失败时返回静态数据  
        return get_news()  
  
NEWS_SCHEMA = {  
    "type": "function",  
    "function": {  
        "name": "get_news",  
        "description": "获取静态财经新闻数据(fallback数据)",  
        "parameters": {  
            "type": "object",  
            "properties": {},  
            "required": []  
        }  
    }  
}  
  
NEWS_AKSHARE_SCHEMA = {  
    "type": "function",  
    "function": {  
        "name": "get_news_from_akshare",  
        "description": "使用AKShare获取东方财富网财经新闻(官方API,最稳定)",  
        "parameters": {  
            "type": "object",  
            "properties": {},  
            "required": []  
        }  
    }  
}

async def main():  
    client = Client(mcp)  
    async with client:  
        # 查看可用工具  
        tools = await client.list_tools()  
        print('=== 可用工具 ===')  
        for tool in tools:  
            print(f"- 工具名: {tool.name}")  
            print(f"  描述: {tool.description[:50]}...\n")  
          
        # 调用静态新闻工具  
        print('=== 静态财经新闻结果 ===')  
        static_result = await client.call_tool('get_news', {})  
        for i, news in enumerate(static_result.structured_content['result'], 1):  
            print(f"{i}. 标题: {news['title']}")  
            print(f"   摘要: {news['summary']}")  
            print(f"   来源: {news['source']} | 日期: {news['date']}\n")  
          
        # 调用AKShare新闻工具  
        print('=== AKShare财经新闻结果 ===')  
        akshare_result = await client.call_tool('get_news_from_akshare', {})  
        for i, news in enumerate(akshare_result.structured_content['result'], 1):  
            if "提示" in news:  
                print(f"{i}. {news['提示']}")  
            else:  
                print(f"{i}. 标题: {news['title']}")  
                print(f"   链接: {news['url']}")  
                print(f"   日期: {news['date']}\n")  
  
# 这是推荐的、唯一的程序入口点  
if __name__ == '__main__':  
    asyncio.run(main())